package spring.cloud.tasks.api.service.impl;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Strings;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.data.Stat;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import spring.cloud.tasks.api.database.entity.TaskConfigurationDto;
import spring.cloud.tasks.api.database.service.TaskConfigurationDtoService;
import spring.cloud.tasks.api.domain.executor.Executor;
import spring.cloud.tasks.api.domain.executor.ExecutorStatus;
import spring.cloud.tasks.api.domain.executor.ExecutorType;
import spring.cloud.tasks.api.domain.task.TaskConfiguration;
import spring.cloud.tasks.api.domain.task.TaskType;
import spring.cloud.tasks.api.domain.task_executor.TaskExecutor;
import spring.cloud.tasks.api.domain.task_executor.TaskExecutorStatus;
import spring.cloud.tasks.api.domain.task_executor.TaskExecutorStatusInfo;
import spring.cloud.tasks.api.domain.task_item.TaskItemExecuteInfo;
import spring.cloud.tasks.api.domain.task_item.TaskItemExecuteStatus;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperClusterService;
import spring.cloud.tasks.api.vo.TaskConfigurationVo;
import spring.cloud.tasks.common.cron_expression.CronExpression;
import spring.cloud.tasks.common.node.*;
import spring.cloud.tasks.common.task.TaskIdAndTaskInstanceIdAndTaskTurnIdAndTaskItemIndex;
import spring.cloud.tasks.common.task.TaskItemExecuteRenewalInfo;
import spring.cloud.tasks.common.zookeeper.ZooKeeper;
import spring.cloud.tasks.common.zookeeper.ZooKeeperRepository;

import javax.annotation.Resource;
import java.util.*;

@Slf4j
public class TaskServiceImpl implements TaskService {


    public static int JOB_CAN_BE_DELETE_TIME_LIMIT = 2 * 60 * 1000;// 任务可以被删除的时间限制(单位：ms)

    @Override
    public int getMaxTaskNum() {
        int DEFAULT_MAX_TASK_NUM = 128;
        return DEFAULT_MAX_TASK_NUM;
    }

    @Override
    public boolean taskIncExceeds(String nameSpace, int maxJobNum, int inc) throws RuntimeException {
        if (maxJobNum <= 0) {
            return false;
        }
        int taskSize = getTaskConfigurationList(nameSpace).size();
        return (taskSize + inc) > maxJobNum;
    }


    /**
     * 修改任务签名。任务签名改变，调度器会进行任务重新分配。
     *
     * @param zooKeeper
     * @param taskId
     */
    private void updateTaskSignature(ZooKeeper zooKeeper, String taskId) {
        String taskConfigSignatureNodePath = TaskNodePath.getTaskConfigSignatureNodePath(taskId);
        zooKeeper.update(taskConfigSignatureNodePath, UUID.randomUUID().toString());
    }

    @Resource
    private ZooKeeperClusterService zooKeeperService;
    @Resource
    private TaskConfigurationDtoService taskConfigurationDtoService;
    @Resource
    private ZooKeeperRepository zooKeeperRepository;

    //
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void addTask(String nameSpace, TaskConfiguration taskConfiguration, String userId) throws RuntimeException {
        validateTaskConfiguration(taskConfiguration);
        // 如果数据存在相同任务名，则抛异常
        String taskId = taskConfiguration.getTaskId();
        if (taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId) != null) {
            throw new RuntimeException(String.format("该任务(%s)在DB已经存在", taskId));
        }
        // 如果zk存在该任务，则尝试删除
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        if (zooKeeper.checkExists(TaskNodePath.getTaskNodePath(taskId))) {
            throw new RuntimeException(String.format("该任务(%s)在ZK已经存在", taskId));
        }
        // 该域任务总数不能超过一定数量
        int maxTaskNum = getMaxTaskNum();
        if (taskIncExceeds(nameSpace, maxTaskNum, 1)) {
            throw new RuntimeException(String.format("总任务数超过最大限制(%d)，任务名%s创建失败", maxTaskNum, taskId));
        }
        // 添加该任务到数据库
        taskConfigurationDtoService.insert(constructTaskConfigurationDto(nameSpace, taskConfiguration, ""));
        // 添加该任务配置到zk
        TaskConfigurationAndZooKeeperUtil.createTaskConfigurationToZooKeeper(zooKeeper, taskConfiguration);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateTaskConfiguration(String nameSpace, TaskConfiguration taskConfiguration) throws RuntimeException {
        TaskConfigurationDto oldTaskConfiguration4Dto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskConfiguration.getTaskId());
        if (oldTaskConfiguration4Dto == null) {
            throw new RuntimeException(String.format("该任务(%s)不存在", taskConfiguration.getTaskId()));
        }
        // 从数据库拿出老的数据，与老的数据库中的该任务的配置对比，如果没有改变，则直接返回
        TaskConfigurationDto newTaskConfigurationDto = new TaskConfigurationDto();
        newTaskConfigurationDto.copyProperties(taskConfiguration);
        if (oldTaskConfiguration4Dto.equals(newTaskConfigurationDto)) {
            return;
        }
        // 校验任务配置
        validateTaskConfiguration(newTaskConfigurationDto);
        // 更新该任务到数据库
        taskConfigurationDtoService.update(newTaskConfigurationDto);
        // 更新任务配置到zk，并联动更新关联任务的上下游
        TaskConfigurationAndZooKeeperUtil.updateTaskConfigurationToZooKeeper(zooKeeperService.connectZooKeeper(nameSpace), newTaskConfigurationDto);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateTaskCronExpression(String nameSpace, String taskId, String cronExpression, String pausePeriodDate, String pausePeriodTime) throws RuntimeException {
        if (cronExpression == null) {
            throw new NullPointerException("cronExpression");
        }
        new CronExpression(cronExpression, pausePeriodDate, pausePeriodTime);
        //
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        if (zooKeeper.checkExists(TaskNodePath.getTaskConfigurationNodePath(taskId))) {
            String cronExpressionNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.cron_expression);
            String pausePeriodDateNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_date);
            String pausePeriodTimeNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_time);
            //
            zooKeeper.update(cronExpressionNodePath, cronExpression);
            zooKeeper.update(pausePeriodDateNodePath, pausePeriodDate);
            zooKeeper.update(pausePeriodTimeNodePath, pausePeriodDate);
        } else {
            throw new RuntimeException("The job does not exists: " + taskId);
        }
    }


    private void validateTaskConfiguration(TaskConfiguration taskConfiguration) throws RuntimeException {
        // 任务类型必填
        if (StringUtils.isBlank(taskConfiguration.getTaskType())) {
            throw new RuntimeException("任务类型必填");
        }
        // 验证任务类型
        TaskType taskType = TaskType.getTaskType(taskConfiguration.getTaskType());
        if (taskType == TaskType.UNKNOWN_TASK) {
            throw new RuntimeException("任务类型未知");
        }
        String xid = taskConfiguration.getXId();
        if (StringUtils.isBlank(xid)) {
            throw new RuntimeException("xid必填");
        }
        // 任务名必填
        String taskId = taskConfiguration.getTaskId();
        if (StringUtils.isBlank(taskId)) {
            throw new RuntimeException("任务名必填");
        }
        // 任务名只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_
        if (!taskId.matches("[0-9a-zA-Z_]*")) {
            throw new RuntimeException("任务名只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_");
        }
        // 校验cron
        if (taskType == TaskType.TASK_TYPE_ITEM_TASK) {
            //校验
        } else if (taskType == TaskType.TASK_TYPE_MESSAGE_QUEUE_TASK) {
            //校验
        } else if (taskType == TaskType.TASK_TYPE_BATCH_TASK) {
            //校验
        }

        validateCronExpression(taskConfiguration);
        validateTaskItem(taskConfiguration);
    }

    private void validateCronExpression(TaskConfiguration taskConfiguration) throws RuntimeException {
        if (TaskType.isCronTaskType(TaskType.getTaskType(taskConfiguration.getTaskType()))) {
            // cron表达式必填
            if (StringUtils.isEmpty(taskConfiguration.getCronExpression())) {
                throw new RuntimeException("对于cron任务，cron表达式必填");
            }
            // cron表达式语法验证
            new CronExpression(taskConfiguration.getCronExpression(), taskConfiguration.getPausePeriodDate(), taskConfiguration.getPausePeriodTime());

        } else {
            taskConfiguration.setCronExpression(""); // 其他类型的不需要持久化保存cron表达式
        }
    }

    private void validateTaskItem(TaskConfiguration taskConfiguration) throws RuntimeException {
        // 分片参数不能小于分片总数
        if (taskConfiguration.getTaskItemTotalCount() == null || taskConfiguration.getTaskItemTotalCount() < 1) {
            throw new RuntimeException("分片数不能为空，并且不能小于1");
        }
        if ((taskConfiguration.getTaskItemTotalCount() > 0) && (
                taskConfiguration.getTaskItemParameters() == null//
                        ||
                        taskConfiguration.getTaskItemParameters().trim().isEmpty()//
                        ||
                        taskConfiguration.getTaskItemParameters().split(",").length < taskConfiguration.getTaskItemTotalCount())) {//
            throw new RuntimeException("分片参数不能小于分片总数");
        }

        String taskItemParametersString = taskConfiguration.getTaskItemParameters();
        String[] taskItemParameters = taskItemParametersString.trim().split(",");
        for (int i = 0; i < taskItemParameters.length; i++) {
            String taskItemParameter = taskItemParameters[i];
            if (!taskItemParameter.contains("=")) {
                throw new RuntimeException(String.format("分片参数'%s'格式有误", taskItemParameter));
            }
            String key = taskItemParameter.trim().split("=")[0].trim();
            boolean isNumeric = StringUtils.isNumeric(key);
            if (!isNumeric) {
                throw new RuntimeException(String.format("分片参数'%s'格式有误", taskConfiguration.getTaskItemParameters()));
            }
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void deleteTask(String nameSpace, String taskId, String userId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能删除该任务（" + taskId + "），因为该任务不存在");
        }
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        Stat stat = zooKeeper.getStat(TaskNodePath.getTaskNodePath(taskId));
        if (stat != null) {
            long createTimeDiff = System.currentTimeMillis() - stat.getCtime();
            if (createTimeDiff < JOB_CAN_BE_DELETE_TIME_LIMIT) {
                throw new RuntimeException(String.format("不能删除该任务(%s)，因为该任务创建时间距离现在不超过%d分钟", taskId, JOB_CAN_BE_DELETE_TIME_LIMIT / 60000));
            }
        }
        // remove job from db
        taskConfigurationDtoService.delete(taskConfigurationDto.getId());
        //同时写入一条异步删除的通知记录
        // remove job from zk
        removeTaskFromZooKeeper(zooKeeper, taskId);
    }

    /**
     * 删除zk上的任务结点。先持久化config/toDelete结点，让executor收到该事件，shutdown自身的该任务。如果所有executor都已经shutdown该任务，则才可以安全删除任务结点。
     *
     * @return 等待executor shutdown任务，等待一定时间后，如果executor还没完全shutdown，则放弃等待，返回false。
     * 否则，在等待时间内，executor都shutdown完全，则删除任务结点，并返回true。
     */
    private boolean removeTaskFromZooKeeper(ZooKeeper zooKeeper, String taskId) throws RuntimeException {
        zooKeeper.update(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled), TaskConfiguration.enabled_n);
        // 1.任务的executor全online的情况，添加toDelete节点，触发监听器动态删除节点
        String toDeleteNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, "toDelete");
        if (zooKeeper.checkExists(toDeleteNodePath)) {
            zooKeeper.deleteRecursive(toDeleteNodePath);
        }
        zooKeeper.create(toDeleteNodePath);
        for (int i = 0; i < 20; i++) {
            // 2.任务的executor全offline的情况，或有几个online，几个offline的情况
            String executorsNodePath = TaskNodePath.getTaskExecutorsNodePath(taskId);
            if (!zooKeeper.checkExists(executorsNodePath)) {
                // (1)如果不存在$Job/taskId/servers节点，说明该任务没有任何executor接管，可直接删除任务节点
                zooKeeper.deleteRecursive(TaskNodePath.getTaskNodePath(taskId));
                return true;
            }
            // (2)如果该任务servers下没有任何executor，可直接删除任务节点
            List<String> executorList = zooKeeper.getChildren(executorsNodePath);
            if (CollectionUtils.isEmpty(executorList)) {
                zooKeeper.deleteRecursive(TaskNodePath.getTaskNodePath(taskId));
                return true;
            }
            // (3)只要该任务没有一个能运行的该任务的executor在线，那么直接删除任务节点
            boolean hasOnlineExecutor = false;
            for (String executor : executorList) {
                if (zooKeeper.checkExists(ExecutorNodePath.getExecutorsNodePath(executor, "ip")) && zooKeeper.checkExists(TaskNodePath.getExecutorStatus(taskId, executor))) {
                    hasOnlineExecutor = true;
                } else {
                    zooKeeper.deleteRecursive(TaskNodePath.getTaskExecutorNodePath(taskId, executor));
                }
            }
            if (!hasOnlineExecutor) {
                zooKeeper.deleteRecursive(TaskNodePath.getTaskNodePath(taskId));
                return true;
            }
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        return false;

    }

    private TaskConfigurationDto constructTaskConfigurationDto(String nameSpace, TaskConfiguration taskConfiguration, String userId) {
        TaskConfigurationDto taskConfigurationDto = new TaskConfigurationDto();
        taskConfigurationDto.copyProperties(taskConfiguration);
        taskConfigurationDto.setNameSpaceKey(nameSpace);
        Date now = new Date();
        taskConfigurationDto.setCreateBy(userId);
        taskConfigurationDto.setLastUpdateBy(userId);
        taskConfigurationDto.setCreateTime(now);
        taskConfigurationDto.setLastUpdateTime(now);
        return taskConfigurationDto;
    }


    @Transactional(rollbackFor = Exception.class)
    @Override
    public void enableTask(String nameSpace, String taskId, String userId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能启用该任务（" + taskId + "），因为该任务不存在");
        }//任务只能禁用不会真正的删除
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        String enabledNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled);
        String enabled = zooKeeper.getData(enabledNodePath);
        if (TaskConfiguration.enabled_y.equals(taskConfigurationDto.getEnabled())) {
            //throw new RuntimeException("该任务（" + taskId + "）已经处于启用状态");
            if (TaskConfiguration.enabled_y.equals(enabled)) {
                throw new RuntimeException("该任务（" + taskId + "）已经处于启用状态");
            } else {
                //zk启用
                enableTask(zooKeeper, taskId);
                //enableTask(taskConfigurationDto, userId);
                return;
            }
        } else {
            if (TaskConfiguration.enabled_y.equals(enabled)) {
                //enableTask(zooKeeper, taskId);
                enableTask(taskConfigurationDto, userId);
                return;
            } else {
                //zk启用
                enableTask(zooKeeper, taskId);
                enableTask(taskConfigurationDto, userId);
                return;
            }
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void disableTask(String nameSpace, String taskId, String userId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能禁用该任务（" + taskId + "），因为该任务不存在");
        }
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        String enabledNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled);
        String enabled = zooKeeper.getData(enabledNodePath);
        if (TaskConfigurationDto.enabled_n.equals(taskConfigurationDto.getEnabled())) {
            if (TaskConfigurationDto.enabled_n.equals(enabled)) {
                throw new RuntimeException("该任务（" + taskId + "）已经处于禁用状态");
            } else {
                disableTask(zooKeeper, taskId);
                //disableTask(taskConfigurationDto, userId);
                return;
            }
        } else {
            if (TaskConfigurationDto.enabled_n.equals(enabled)) {
                //disableTask(zooKeeper, taskId);
                disableTask(taskConfigurationDto, userId);
                return;
            } else {
                disableTask(zooKeeper, taskId);
                disableTask(taskConfigurationDto, userId);
                return;
            }

        }
    }

    private void enableTask(TaskConfigurationDto taskConfigurationDto, String userId) {
        taskConfigurationDto.setEnabled(TaskConfiguration.enabled_y);
        taskConfigurationDto.setLastUpdateBy(userId);
        taskConfigurationDto.setLastUpdateTime(new Date());
        taskConfigurationDtoService.update(taskConfigurationDto);
    }

    private void disableTask(TaskConfigurationDto taskConfigurationDto, String userId) {
        taskConfigurationDto.setEnabled(TaskConfigurationDto.enabled_n);
        taskConfigurationDto.setLastUpdateBy(userId);
        taskConfigurationDto.setLastUpdateTime(new Date());
        taskConfigurationDtoService.update(taskConfigurationDto);
    }

    private void enableTask(ZooKeeper zooKeeper, String taskId) {
        zooKeeper.update(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled), TaskConfiguration.enabled_y);
    }

    private void disableTask(ZooKeeper zooKeeper, String taskId) {
        zooKeeper.update(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled), TaskConfigurationDto.enabled_n);
    }


    @Override
    public void runAtOnce(final String nameSpace, final String taskId, final Map<String, Object> triggeredData) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        List<TaskExecutorStatusInfo> taskExecutorStatusInfoList = getTaskExecutorStatusInfoList(nameSpace, taskId);
        if (CollectionUtils.isEmpty(taskExecutorStatusInfoList)) {
            throw new RuntimeException(String.format("没有executor接管该任务(%s)，不能立即启动", taskId));
        }
        boolean hasOnlineExecutor = false;
        for (TaskExecutorStatusInfo taskExecutorStatusInfo : taskExecutorStatusInfoList) {
            if (TaskExecutorStatus.ONLINE.equals(taskExecutorStatusInfo.getStatus())) {
                hasOnlineExecutor = true;
                log.info("runAtOnce nameSpace:{}, taskId:{}, executorName:{}, triggeredData:{}", nameSpace, taskId);
            }
        }

        if (!hasOnlineExecutor) {
            throw new RuntimeException(String.format("没有在线的executor接管该任务(%s)，不能立即启动", taskId));
        }
        String path = TaskNodePath.getRunOneTimePath(taskId);
        if (zooKeeper.checkExists(path)) {
            zooKeeper.delete(path);
        }
        String triggeredDataStr = JSON.toJSONString(triggeredData);
        zooKeeper.create(path, triggeredDataStr);
    }

    @Override
    public void stopAtOnce(String nameSpace, String taskId, final Map<String, Object> triggeredData) throws RuntimeException {
        List<String> executorIdList = getExecutorIdList(nameSpace, taskId);
        if (!(executorIdList != null && !executorIdList.isEmpty())) {
            throw new RuntimeException(String.format("没有executor接管该任务(%s)，不能立即终止", taskId));
        } else {
            ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
            String path = TaskNodePath.getStopOneTimePath(taskId);
            if (zooKeeper.checkExists(path)) {
                zooKeeper.delete(path);
            }
            String triggeredDataStr = JSON.toJSONString(triggeredData);
            zooKeeper.create(path, triggeredDataStr);
        }
    }

    @Override
    public TaskConfigurationVo getTaskConfigurationVo(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException(String.format("该任务(%s)不存在", taskId));
        }
        TaskConfiguration taskConfiguration = new TaskConfiguration();
        taskConfiguration.copyProperties(taskConfigurationDto);
        TaskConfigurationVo taskConfigurationVo = new TaskConfigurationVo();
        taskConfigurationVo.copyFrom(taskConfiguration);
        taskConfigurationVo.setExecutorList(getExecutorList(nameSpace, taskId));
        return taskConfigurationVo;
    }


    @Override
    public List<String> getTaskIdList(String nameSpace) throws RuntimeException {
        List<String> taskIdList = taskConfigurationDtoService.getTaskIdListByNameSpace(nameSpace);
        return taskIdList != null ? taskIdList : new ArrayList<>();
    }

    @Override
    public List<TaskConfiguration> getTaskConfigurationList(String nameSpace) throws RuntimeException {
        List<TaskConfiguration> taskConfigurationList = new ArrayList<>();
        List<TaskConfigurationDto> taskConfigurationDtoList = taskConfigurationDtoService.getTaskConfigurationDtoListByNameSpace(nameSpace);
        if (taskConfigurationDtoList != null) {
            for (TaskConfigurationDto taskConfigurationDto : taskConfigurationDtoList) {
                TaskConfiguration taskConfiguration = new TaskConfiguration();
                taskConfiguration.copyProperties(taskConfigurationDto);
                taskConfigurationList.add(taskConfiguration);
            }
        }
        return taskConfigurationList;
    }

    @Override
    public TaskConfiguration getTaskConfiguration(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException(String.format("该任务(%s)不存在", taskId));
        }
        TaskConfiguration taskConfiguration = new TaskConfiguration();
        taskConfiguration.copyProperties(taskConfigurationDto);
        return taskConfiguration;
    }

    @Override
    public List<String> getTaskIdListFromZooKeeper(String nameSpace) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        String tasksNodePath = TaskNodePath.getTasksNodePath();
        List<String> taskIdList = zooKeeper.getChildren(tasksNodePath);
        if (taskIdList == null) {
            return new ArrayList<>();
        }
        List<String> $taskIdList = new ArrayList<>();
        for (String taskId : taskIdList) {
            // 如果config节点存在才视为正常任务，其他异常任务在其他功能操作时也忽略
            if (zooKeeper.checkExists(TaskNodePath.getTaskConfigurationNodePath(taskId))) {
                $taskIdList.add(taskId);
            }
        }
        Collections.sort($taskIdList);
        return $taskIdList;
    }

    @Override
    public List<TaskConfiguration> getTaskInfoList(final String nameSpace) throws RuntimeException {
        List<TaskConfiguration> taskInfoList = new ArrayList<>();
        List<TaskConfiguration> taskConfigurationList = getTaskConfigurationList(nameSpace);
        if (taskConfigurationList != null) {
            for (TaskConfiguration taskConfiguration : taskConfigurationList) {
                try {
                    TaskConfiguration taskInfo = getTaskInfo(nameSpace, taskConfiguration.getTaskId());
                    taskInfoList.add(taskInfo);
                } catch (Exception e) {
                    log.error("getRestApiJobInfos exception:", e);
                    continue;
                }
            }
        }
        return taskInfoList;
    }


    @Override
    public TaskConfiguration getTaskInfo(String nameSpace, final String taskId) throws RuntimeException {
        TaskConfiguration taskInfo = construct(zooKeeperService.connectZooKeeper(nameSpace), taskId);
        return taskInfo;
    }


    private TaskConfiguration construct(ZooKeeper zooKeeper, String taskId) {
        // 设置任务配置信息
        TaskConfiguration taskConfiguration = TaskConfigurationAndZooKeeperUtil.getTaskConfigurationFromZooKeeper(zooKeeper, taskId);
        return taskConfiguration;
    }


    @Override
    public List<TaskItemExecuteInfo> getTaskItemExecuteInfoList(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        TaskConfiguration taskConfiguration = getTaskConfiguration(nameSpace, taskId);
        if (!taskConfiguration.getEnabled().equals("Y")) {
            return new ArrayList<>();
        }
        // 如果execution节点不存在则返回空List
        List<TaskIdAndTaskInstanceIdAndTaskTurnIdAndTaskItemIndex> taskInstanceAndTaskItemList = BlockNodePaths.getTaskIdAndTaskInstanceIdAndTaskTurnIdAndTaskItemIndexList(zooKeeper, taskId);
        List<TaskItemExecuteInfo> taskItemExecuteInfoList = new ArrayList<>();
        for (TaskIdAndTaskInstanceIdAndTaskTurnIdAndTaskItemIndex taskInstanceAndTaskItem : taskInstanceAndTaskItemList) {
            TaskItemExecuteInfo taskItemExecuteInfo = getTaskItemExecuteInfo(zooKeeper, taskInstanceAndTaskItem);
            taskItemExecuteInfoList.add(taskItemExecuteInfo);
        }
        Collections.sort(taskItemExecuteInfoList);

        return taskItemExecuteInfoList;
    }


    private TaskItemExecuteInfo getTaskItemExecuteInfo(ZooKeeper zooKeeper, TaskIdAndTaskInstanceIdAndTaskTurnIdAndTaskItemIndex taskInstanceAndTaskItem) {
        TaskItemExecuteInfo taskItemExecuteInfo = new TaskItemExecuteInfo();
        taskItemExecuteInfo.setTaskId(taskInstanceAndTaskItem.getTaskId());
        taskItemExecuteInfo.setTaskInstanceId(taskInstanceAndTaskItem.getTaskInstanceId());
        taskItemExecuteInfo.setTaskItemIndex(taskInstanceAndTaskItem.getTaskItemIndex());
        TaskItemExecuteRenewalInfo taskItemExecuteRenewalInfo = BlockNodePaths.getTaskItemExecuteRenewalInfo(zooKeeper, taskInstanceAndTaskItem);
        if (taskItemExecuteRenewalInfo == null) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.NULL);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.PENDING == taskItemExecuteRenewalInfo.getCode()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.PENDING);
            return taskItemExecuteInfo;
        }
        taskItemExecuteInfo.setExecutorId(taskItemExecuteRenewalInfo.getExecutorId());
        taskItemExecuteInfo.setLastBeginTime(taskItemExecuteRenewalInfo.getExecutorExecuteTime());
        taskItemExecuteInfo.setExecutornReportTime(taskItemExecuteRenewalInfo.getExecutornReportTime());
        taskItemExecuteInfo.setTimeConsumed((taskItemExecuteRenewalInfo.getExecutornReportTime() - taskItemExecuteRenewalInfo.getExecutorExecuteTime()) / 1000);
        //
        if (TaskItemExecuteStatus.COMPLETED == taskItemExecuteRenewalInfo.getCode()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.COMPLETED);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.FAILED == taskItemExecuteRenewalInfo.getCode()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.FAILED);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.TIMEOUT == taskItemExecuteRenewalInfo.getCode()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.TIMEOUT);
            return taskItemExecuteInfo;
        }
        if (TaskItemExecuteStatus.RUNNING == taskItemExecuteRenewalInfo.getCode()) {
            taskItemExecuteInfo.setStatus(TaskItemExecuteStatus.RUNNING);
            return taskItemExecuteInfo;
        }
        throw new IllegalStateException("暂时还不支持其他的状态");
    }


    /////////////////////////////////////////////////////////////////////////////////
    @Override
    public List<Executor> getExecutorList(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能获取该任务（" + taskId + "）可选择的优先Executor，因为该任务不存在");
        }
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        String taskNodePath = TaskNodePath.getTaskNodePath(taskId);
        if (!zooKeeper.checkExists(taskNodePath)) {
            return null;
        }
        List<Executor> executorList = new ArrayList<>();
        String executorsNodePath = ExecutorNodePath.getExecutorsNodePath();
        if (!zooKeeper.checkExists(executorsNodePath)) {
            return executorList;
        }
        List<String> executorIdList = zooKeeper.getChildren(executorsNodePath);
        if (executorIdList == null) {
            executorIdList = new ArrayList<>();
        }
        if (!executorIdList.isEmpty()) {
            for (String executorId : executorIdList) {
                if (!executorId.startsWith("@")) {
                    Executor executor = new Executor();
                    executor.setExecutorType(ExecutorType.PHYSICAL_MACHINE);
                    executor.setExecutorId(executorId);
                    executor.setOwnTraffic(zooKeeper.checkExists(ExecutorNodePath.getExecutorOwnTrafficNodePath(executorId)));
                    String ip = zooKeeper.getData(ExecutorNodePath.getExecutorIpNodePath(executorId));
                    if (StringUtils.isNotBlank(ip)) {
                        executor.setExecutorStatus(ExecutorStatus.ONLINE);
                        executor.setIp(ip);
                    } else {
                        executor.setExecutorStatus(ExecutorStatus.OFFLINE);
                    }
                    executorList.add(executor);
                } else {
                    Executor executor = new Executor();
                    executor.setExecutorType(ExecutorType.CONTAINER);
                    executor.setExecutorId(executorId);
                    executor.setOwnTraffic(zooKeeper.checkExists(ExecutorNodePath.getExecutorOwnTrafficNodePath(executorId)));
                    String ip = zooKeeper.getData(ExecutorNodePath.getExecutorIpNodePath(executorId));
                    if (StringUtils.isNotBlank(ip)) {
                        executor.setExecutorStatus(ExecutorStatus.ONLINE);
                        executor.setIp(ip);
                    } else {
                        executor.setExecutorStatus(ExecutorStatus.OFFLINE);
                    }
                    executorList.add(executor);
                }

            }
        }
        String preferExecutorIdListNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list);
        if (!zooKeeper.checkExists(preferExecutorIdListNodePath)) {
            return executorList;
        }
        String preferExecutorIdListSting = zooKeeper.getData(preferExecutorIdListNodePath);
        if (Strings.isNullOrEmpty(preferExecutorIdListSting)) {
            return executorList;
        }
        String[] preferExecutorIdList = preferExecutorIdListSting.split(",");
        for (String preferExecutorId : preferExecutorIdList) {
            if (!preferExecutorId.startsWith("@")) {
                boolean include = false;
                for (Executor executor : executorList) {
                    if (executor.getExecutorId().equals(preferExecutorId)) {
                        include = true;
                        break;
                    }
                }
                if (!include) {
                    Executor executor = new Executor();
                    executor.setExecutorId(preferExecutorId);
                    executor.setExecutorType(ExecutorType.PHYSICAL_MACHINE);
                    executor.setExecutorStatus(ExecutorStatus.DELETED);
                    executorList.add(executor);
                }
            } else {
                boolean include = false;
                for (Executor executor : executorList) {
                    if (executor.getExecutorId().equals(preferExecutorId)) {
                        include = true;
                        break;
                    }
                }
                if (!include) {
                    Executor executor = new Executor();
                    executor.setExecutorId(preferExecutorId);
                    executor.setExecutorType(ExecutorType.CONTAINER);
                    executor.setExecutorStatus(ExecutorStatus.DELETED);
                    executorList.add(executor);
                }
            }
        }
        return executorList;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void setPreferExecutorIdList(String nameSpace, String taskId, String preferExecutorIdListSting) throws RuntimeException {
        TaskConfigurationDto oldTaskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (oldTaskConfigurationDto == null) {
            throw new RuntimeException("设置该任务（" + taskId + "）优先Executor失败，因为该任务不存在");
        }
        oldTaskConfigurationDto.setPreferExecutorIdList(preferExecutorIdListSting);
        taskConfigurationDtoService.updatePreferExecutorIdList(oldTaskConfigurationDto);
        // save to zk
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        String taskConfigPreferListNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list);
        zooKeeper.update(taskConfigPreferListNodePath, preferExecutorIdListSting);
        // delete and create the forceShard node
        String taskConfigForceShardNodePath = TaskNodePath.getTaskConfigForceShardNodePath(taskId);
        zooKeeper.delete(taskConfigForceShardNodePath);
        zooKeeper.create(taskConfigForceShardNodePath);
        //
        updateTaskSignature(zooKeeper, taskId);
    }


    @Override
    public List<TaskExecutor> getTaskExecutorList(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        String executorsNodePath = TaskNodePath.getTaskExecutorsNodePath(taskId);
        List<String> executorIdList = zooKeeper.getChildren(executorsNodePath);
        List<TaskExecutor> taskExecutorList = new ArrayList<>();
        if (executorIdList != null && !executorIdList.isEmpty()) {
            for (String executorId : executorIdList) {
                TaskExecutor taskExecutor = getTaskExecutor(zooKeeper, taskId, executorId);
                taskExecutorList.add(taskExecutor);
            }
        }
        return taskExecutorList;
    }

    private TaskExecutor getTaskExecutor(ZooKeeper zooKeeper, String taskId, String executorId) {
        TaskExecutor taskExecutor = new TaskExecutor();
        taskExecutor.setTaskId(taskId);
        taskExecutor.setExecutorId(executorId);
        taskExecutor.setContainer(executorId.startsWith("@"));
        taskExecutor.setIp(zooKeeper.getData(TaskNodePath.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.ip)));
        taskExecutor.setStatus(getTaskExecutorStatus(zooKeeper, taskId, executorId));
        //
        taskExecutor.setTaskItemAssign(zooKeeper.getData(TaskNodePath.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.TASK_ITEM_ASSIGN)));
        //
        String processSuccessCount = zooKeeper.getData(TaskNodePath.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.processSuccessCount));
        taskExecutor.setProcessSuccessCount(null == processSuccessCount ? 0 : Integer.parseInt(processSuccessCount));
        String processFailureCount = zooKeeper.getData(TaskNodePath.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.processFailureCount));
        taskExecutor.setProcessFailureCount(null == processFailureCount ? 0 : Integer.parseInt(processFailureCount));
        //
        return taskExecutor;
    }

    @Override
    public List<TaskExecutorStatusInfo> getTaskExecutorStatusInfoList(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        List<TaskExecutorStatusInfo> taskExecutorStatusInfoList = new ArrayList<>();
        List<String> executorIdList = getExecutorIdList(nameSpace, taskId);
        if (executorIdList != null) {
            for (String executorId : executorIdList) {
                taskExecutorStatusInfoList.add(getTaskExecutorStatusInfo(zooKeeper, taskId, executorId));
            }
        }
        return taskExecutorStatusInfoList;
    }

    @Override
    public List<String> getExecutorIdList(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        String taskNodePath = TaskNodePath.getTaskNodePath(taskId);
        if (!zooKeeper.checkExists(taskNodePath)) {
            return null;
        }
        String executorsNodePath = TaskNodePath.getTaskExecutorsNodePath(taskId);
        List<String> executorIdList = zooKeeper.getChildren(executorsNodePath);
        if (executorIdList == null || CollectionUtils.isEmpty(executorIdList)) {
            return new ArrayList<>();
        }
        return executorIdList;
    }

    private TaskExecutorStatusInfo getTaskExecutorStatusInfo(ZooKeeper zooKeeper, String taskId, String executorId) {
        TaskExecutorStatusInfo taskExecutorStatus = new TaskExecutorStatusInfo();
        taskExecutorStatus.setTaskId(taskId);
        taskExecutorStatus.setExecutorId(executorId);
        taskExecutorStatus.setStatus(getTaskExecutorStatus(zooKeeper, taskId, executorId));
        return taskExecutorStatus;
    }

    private String getTaskExecutorStatus(ZooKeeper zooKeeper, String taskId, String executorId) {
        return zooKeeper.getData(TaskNodePath.getTaskExecutorNodePath(taskId, executorId, TaskExecutorNode.status));
    }


}

